package com.chen.mqtthandle.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.core.MessageProducer;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.handler.LoggingHandler;
import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;

public class MqttConfig {
    @Bean(value = "clientFactory")
    public DefaultMqttPahoClientFactory defaultMqttPachoClientFactory(){
        DefaultMqttPahoClientFactory defaultMqttPahoClientFactory = new DefaultMqttPahoClientFactory();
        defaultMqttPahoClientFactory.setUserName("");
        defaultMqttPahoClientFactory.setPassword("");
        defaultMqttPahoClientFactory.setServerURIs("");
        return defaultMqttPahoClientFactory;
    }
    @Bean(value = "mqtt")
    public MqttPahoMessageHandler mqttPahoMessageHandler(){
        MqttPahoMessageHandler mqttPahoMessageHandler = new MqttPahoMessageHandler("",defaultMqttPachoClientFactory());
        return mqttPahoMessageHandler;
    }

    @Bean
    public IntegrationFlow mqttInFlow(){
        return IntegrationFlows.from(mqttInbound()).transform(p -> p+",received from MQTT")
                .handle(logger())
                .get();
    }

    @Bean
    public LoggingHandler logger(){
        LoggingHandler loggingHandler = new LoggingHandler("INFO");
        loggingHandler.setLoggerName("siSimple");
        return loggingHandler;
    }

    @Bean
    public MessageProducerSupport mqttInbound(){
        MqttPahoMessageDrivenChannelAdapter adater = new MqttPahoMessageDrivenChannelAdapter("siSampleConsumer",
                defaultMqttPachoClientFactory(),"siSampleTopic");
        adater.setCompletionTimeout(5000);
        adater.setConverter(new DefaultPahoMessageConverter());
        adater.setQos(1);
        return adater;
    }

    @Bean
    public IntegrationFlow mqttOutFlow() {
        //console input
//        return IntegrationFlows.from(CharacterStreamReadingMessageSource.stdin(),
//                e -> e.poller(Pollers.fixedDelay(1000)))
//                .transform(p -> p + " sent to MQTT")
//                .handle(mqttOutbound())
//                .get();
        return IntegrationFlows.from(outChannel())
                .handle(mqttOutbound())
                .get();
    }

    @Bean
    public MessageChannel outChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler("siSamplePublisher", defaultMqttPachoClientFactory());
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic("siSampleTopic");
        return messageHandler;
    }

}
